package com.zyx.flinkdemo.stream.window.trigger;

import com.zyx.flinkdemo.pojo.WordCountTs;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeoutTrigger;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/3 10:43
 * desc: ProcessingTimeoutTrigger案例
 */
public class ProcessingTimeoutTriggerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WordCountTs> wordCountDs = env
                .socketTextStream("localhost", 7777)
                .map(new MapFunction<String, WordCountTs>() {
                    @Override
                    public WordCountTs map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WordCountTs(split[0],  Integer.parseInt(split[1]), split[2], Long.parseLong(split[3]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WordCountTs>forMonotonousTimestamps()
                        .withTimestampAssigner((SerializableTimestampAssigner<WordCountTs>) (element, recordTimestamp) -> element.getTs()));


        wordCountDs
                .keyBy(WordCountTs::getWord)
                // 开间隔为5s的滚动窗口
                .window(TumblingEventTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
                // 使用ProcessingTimeoutTrigger作为触发器, 内部的触发器使用EventTimeTrigger, 超时时间为8s
                .trigger(ProcessingTimeoutTrigger.of(EventTimeTrigger.create(), Duration.ofSeconds(8L)))
                .reduce(new ReduceFunction<WordCountTs>() {
                    @Override
                    public WordCountTs reduce(WordCountTs value1, WordCountTs value2) throws Exception {
                        return new WordCountTs(value1.getWord(),
                                value1.getCount() + value2.getCount(),
                                value1.getTime().compareTo(value2.getTime()) >= 0 ? value1.getTime() : value2.getTime(),
                                Math.max(value1.getTs(), value2.getTs()));
                    }
                })
                .print();

        env.execute();
    }
}
